package thirdpart.tcp;

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetSocket;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;

import java.awt.print.Pageable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Author: Tzu
 * @Date: 2020/11/18 19:26
 * @Version: 1.0
 * @Description: 发送Tcp报文
 */

@Log4j2
@RequiredArgsConstructor
public class TcpDirectSender {

    @NonNull
    private String ip;

    @NonNull
    private int port;

    @NonNull
    private Vertx vertx;

    ////////////////////////////////////////////////////////////////

    //volatile-->取内存中最新创建的socket
    //防止断连后拿到旧的socket进行连接操作
    private volatile NetSocket socket;

    public  void startup(){
        //连接远程服务器
        vertx.createNetClient().connect(port,ip,new ClientConnHandler());
        //开启一个线程真正去发送tcp数据流
        //只要缓存中有数据队列中就会被取出发送出去
        new Thread(() -> {
            while (true) {
                try {
                    //poll-->超过5s没有获取到数据返回一个null
                    Buffer msgBuffer = sendCache.poll(5, TimeUnit.SECONDS);
                    if (msgBuffer != null
                          && msgBuffer.length() > 0
                          && socket != null) {
                        //发送数据
                        socket.write(msgBuffer);
                    }
                }catch (Exception e){
                    log.error("get buffer from BlockingQueue error : "+e.getMessage());
                }
            }
        }).start();
    }

    //异步处理缓存队列
    private final BlockingQueue<Buffer> sendCache = new LinkedBlockingQueue<>();

    //暴露发送接口： 外部调用send==>实际是把数据存入缓存队列中
    public boolean send(Buffer bufferMsg){
        return sendCache.offer(bufferMsg);
    }


    private class ClientConnHandler implements Handler<AsyncResult<NetSocket>>{

        //重连方法
        private void  reconnect(){
            //每五秒执行一次重连
            vertx.setTimer(1000*5,res -> {
                log.info("try reconnect ro server to {}:{} failed",ip,port);
                vertx.createNetClient().connect(port,ip,new ClientConnHandler());
            });
        }

        @Override
        public void handle(AsyncResult<NetSocket> result) {
            if (result.succeeded()){
                log.info("connect success to remote {}:{}",ip,port);
                socket = result.result();

                //连接断开处理器
                socket.closeHandler(closr -> {
                   log.info("connect to remote {} close",socket.remoteAddress());
                   //重连
                   reconnect();
                });

                //连接异常处理器
                socket.exceptionHandler(ex -> {
                    log.error("error exit",ex.getCause());
                });
            }else {
                log.info("connect failed to remote {}:{}",ip,port);
                //重连
                reconnect();
            }
        }
    }



}
